﻿using log4net.Appender;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using log4net.Core;
using KafkaNet;
using System.IO;
using KafkaNet.Model;

namespace log4net.Kafka
{
    public class KafkaBufferingAppender : BufferingAppenderSkeleton
    {

        private Producer _producer;

        public KafkaSettings KafkaSettings { get; set; }

        public override void ActivateOptions()
        {
            base.ActivateOptions();
            try
            {
                if (KafkaSettings == null) throw new LogException("KafkaSettings is missing");

                if (KafkaSettings.Brokers == null || KafkaSettings.Brokers.Count == 0) throw new Exception("Broker is not found");

                if (_producer == null)
                {
                    var brokers = KafkaSettings.Brokers.Select(x => new Uri(x)).ToArray();
                    var kafkaOptions = new KafkaOptions(brokers);
#if DEBUG
                    kafkaOptions.Log = new ConsoleLog();
#else
                    kafkaOptions.Log = new KafkaLog();
#endif
                    _producer = new Producer(new BrokerRouter(kafkaOptions));
                }
            }
            catch (Exception ex)
            {
                ErrorHandler.Error("could not stop producer", ex);
            }

        }
        private string GetTopic(LoggingEvent loggingEvent)
        {

            string topic = null;
            if (KafkaSettings.Topic != null)
            {
                var sb = new StringBuilder();
                using (var sw = new StringWriter(sb))
                {
                    KafkaSettings.Topic.Format(sw, loggingEvent);
                    topic = sw.ToString();
                }
            }

            if (string.IsNullOrEmpty(topic))
            {
                topic = $"{loggingEvent.LoggerName}.{loggingEvent.Level.Name}";
            }

            return topic;
        }
        private string GetMessage(LoggingEvent loggingEvent)
        {
            var sb = new StringBuilder();
            using (var sr = new StringWriter(sb))
            {
                Layout.Format(sr, loggingEvent);

                return sr.ToString();
            }
        }

        protected override void OnClose()
        {
            base.OnClose();

            try
            {
                _producer?.Stop();
            }
            catch (Exception ex)
            {
                ErrorHandler.Error("could not start producer", ex);
            }

        }

        protected override async void SendBuffer(LoggingEvent[] events)
        {

            foreach (var loggingEvent in events)
            {

                var topic = GetTopic(loggingEvent);
                var message = GetMessage(loggingEvent);

                await _producer.SendMessageAsync(topic, new[] { new KafkaNet.Protocol.Message(message) });

            }

        }
    }
}
